package com.atguigu.gmall.function;

import com.atguigu.gmall.util.JdbcUtil;
import com.atguigu.gmall.util.RedisUtil;
import com.atguigu.gmall.util.ThreadUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/2 14:09
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    
    private ThreadPoolExecutor threadPool;
    private Connection phoenixConn;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadUtil.getThreadPool();
        
        phoenixConn = JdbcUtil.getPhoenixConnection();
    }
    
    @Override
    public void close() throws Exception {
        if (phoenixConn != null) {
            phoenixConn.close();
        }
    
        if (threadPool != null) {
            threadPool.shutdown();
        }
    }
    
    public abstract void addDim(Connection phoenixConn,
                                Jedis  client,
                                T input,
                                ResultFuture<T> resultFuture) throws Exception;
    
    // 只需要实现这个方法就可
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        // 客户端需要有支持异步的api, 如果没有, 则可以使用多线程来完成
        threadPool.submit(() -> {
            // 读取维度操作放在这里就可以了
            // redis在异步使用的时候, 必须每个操作单独得到一个客户端
    
            Jedis client = RedisUtil.getRedisClient();
    
            // 和具体的业务相关
            try {
                addDim(phoenixConn, client, input, resultFuture);
            } catch (Exception e) {
                e.printStackTrace();
                throw new RuntimeException("异步执行的时候的异常, " +
                                               "请检测异步操作: Phoenix是否开启, redis是否开启, hadoop是否开启, maxwell是否开启 ....");
            }
    
            client.close();
    
        });
    }
    
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        // 超时的时候回调这个方法
    
        System.out.println("超时: " + input);
    }
}
